<?php

namespace App\Kuafu\Shop\Command;

use App\Kuafu\Shop\Model\CameraModel;
use App\Kuafu\Shop\Model\ShopMonitorDataModel;
use App\Library\FeiShu\SendFs;
use Illuminate\Console\Command;
use Stomp\Client;
use Stomp\Exception\StompException;
use Stomp\Network\Observer\Exception\HeartbeatException;
use Stomp\Network\Observer\ServerAliveObserver;
use Stomp\StatefulStomp;

class AliyunIotAMQTCommand extends Command
{
    protected $signature = 'aliyunIot:amqp';
    protected $hidden    = true;

    protected int $insertIntervalTime  = 1800; // 入库 间隔 1800 秒
    protected int $insertIntervalCount = 200; // 入库 间隔 条数

    protected array $equipmentList = []; // 设备列表

    protected array $insertData = []; // 待入库的数据列表

    private string $consumerGroupId = '1RtMsWptiUgGheG2yU16000100';
    private string $amqpDomain      = 'ssl://iot-060a0b6k.amqp.iothub.aliyuncs.com:61614';

    public function handle()
    {
        // 线上服务消费数据
        if (config('app.env') != 'prod') {
            return;
        }

        $this->init();

        $accessKey       = config('camera.aliyun.aliyunCameraAccessId');
        $accessSecret    = config('camera.aliyun.aliyunCameraAccessSecret');
        $consumerGroupId = $this->consumerGroupId;
        $clientId        = date('Ymd');
        $iotInstanceId   = config('camera.aliyun.iotInstanceId');
        $timeStamp       = round(microtime(true) * 1000);
        $signMethod      = "hmacsha1";

        $userName    = $clientId . "|authMode=aksign"
            . ",signMethod=" . $signMethod
            . ",timestamp=" . $timeStamp
            . ",authId=" . $accessKey
            . ",iotInstanceId=" . $iotInstanceId
            . ",consumerGroupId=" . $consumerGroupId
            . "|";
        $signContent = "authId=" . $accessKey . "&timestamp=" . $timeStamp;
        $password    = base64_encode(hash_hmac("sha1", $signContent, $accessSecret, $raw_output = true));
        $client      = new Client($this->amqpDomain);

        $sslContext = ['ssl' => ['verify_peer' => true, 'verify_peer_name' => false],];
        $client->getConnection()->setContext($sslContext);

        $observer = new ServerAliveObserver();
        $client->getConnection()->getObservers()->addObserver($observer);

        $client->setHeartbeat(0, 30000);
        $client->setLogin($userName, $password);
        try {
            $client->connect();
        } catch (StompException $e) {
            $this->error("failed to connect to server, msg:" . $e->getMessage());
            SendFs::newRobot()->setTitleContent('failed to connect to server', $e->getMessage(), '3fgc8798')->send();
        }

        $stomp = new StatefulStomp($client);
        $stomp->subscribe('/topic/#');
        $this->line("[" . date('Y-m-d H:i:s') . "] connect success");
        SendFs::newRobot(SendFs::ROBOT_BUSINESS_WARING)
              ->setTitleContent('amqp connect success', 'amqp connect success', '3fgc8798')->send();

        $insertTime = time();
        while (true) {
            try {
                // 检查连接状态
                if (!$client->isConnected()) {
                    $this->warn("connection not exists, will reconnect after 10s.");
                    sleep(10);
                    $client->connect();
                    $stomp->subscribe('/topic/#');
                    $this->line("connect success");
                }

                //处理消息业务逻辑。
                $message = $stomp->read();

                if (!empty($message) && !empty($message->getBody())) {
                    $data = json_decode($message->getBody(), 1);

                    // 处理逻辑
                    $type = $data['type'] ?? '';
                    switch ($type) {
                        case 'alert': // 智能告警
                            $eventList = !empty($data['value']['EventList']) ? $data['value']['EventList'] : [];
                            foreach ($eventList as $eventInfo) {
                                switch ($eventInfo['AlarmType'] ?? 0) {
                                    case 14001: // 14001 客流数据
                                        $eventData = json_decode($eventInfo['Data'], 1);
                                        if (empty($eventData)) {
                                            break;
                                        }

                                        $equipmentInfo = $this->getEquipmentInfo($eventInfo['DeviceName'],
                                            $eventInfo['ProductKey']);

                                        if (empty($equipmentInfo)) {
//                                            $this->line(vsprintf('Skip camera info: {dn:%s,pk:%s}',
//                                                [$eventInfo['DeviceName'], $eventInfo['ProductKey']]));
                                            break;
                                        }

                                        $this->insertData[] = [
                                            'shopId'       => $equipmentInfo['shopId'],
                                            'monitorId'    => $equipmentInfo['id'],
                                            'type'         => $this->switchType($eventData['ActionType']),
                                            'dataAt'       => (int)($eventInfo['EventTime'] / 1000),
                                            'count'        => 1,
                                            'temperature'  => 0,
                                            'createdAt'    => time(),
                                            'updatedAt'    => time(),
                                            'createdByUid' => 0,
                                            'updatedByUid' => 0,
                                        ];

                                        if (count($this->insertData) >= $this->insertIntervalCount) {
                                            ShopMonitorDataModel::query()->insert($this->insertData);
                                            $this->info(vsprintf('[%s] 触发条数 入库 %s 条',
                                                [date('Y-m-d H:i:s'), count($this->insertData)]));
                                            $this->insertData = []; // 重置
                                            $insertTime       = time(); // 重置
                                        }

                                        break;
                                    default:
                                        break;
                                }
                            }

                            break;
                        default:
                            break;
                    }
                }

                // 时间到了入库
                if ($insertTime + $this->insertIntervalTime <= time()) {
                    ShopMonitorDataModel::query()->insert($this->insertData);
                    $this->info(vsprintf('[%s] 触发时间 入库 %s 条',
                        [date('Y-m-d H:i:s'), count($this->insertData)]));

                    $this->insertData = []; // 重置
                    $insertTime       = time(); // 重置
                }

//                $this->line('Skip Loop.');
            } catch (HeartbeatException $e) {
                SendFs::newRobot()
                      ->setTitleContent('The server failed to send us heartbeats within the defined interval', '',
                          '3fgc8798')
                      ->send();

                $this->error('The server failed to send us heartbeats within the defined interval.');

                $stomp->getClient()->disconnect();
            } catch (\Exception $e) {
                $this->error('process message occurs error: ' . $e->getMessage());

                SendFs::newRobot()
                      ->setTitleContent('process message occurs error:', $e->getMessage(),
                          '3fgc8798')
                      ->send();

                $stomp->getClient()->disconnect();
            }
        }
    }

    protected function getEquipmentInfo(string $dn, string $pk): ?object
    {
        $key = $dn . '-' . $pk;
        if (empty($this->equipmentList[$key])) {
            $info = CameraModel::query()
                               ->where('deviceName', $dn)
                               ->where('productKey', $pk)
                               ->first();

            if (empty($info)) {
                return null;
            }

            $this->equipmentList[$key] = $info;
        }
        return $this->equipmentList[$key] ?? null;
    }

    protected function switchType(int $eventAction): int
    {
        return $eventAction + 1; // 正好 +1 能对应上
    }

    protected function init()
    {
        declare(ticks=1);

        pcntl_signal(SIGTERM, function () {
            ShopMonitorDataModel::query()->insert($this->insertData);
            $this->info(vsprintf('[%s] 触发SIGTERM 入库 %s 条',
                [date('Y-m-d H:i:s'), count($this->insertData)]));
            exit;
        });

        pcntl_signal(SIGHUP, function () {
            ShopMonitorDataModel::query()->insert($this->insertData);
            $this->info(vsprintf('[%s] 触发SIGHUP 入库 %s 条',
                [date('Y-m-d H:i:s'), count($this->insertData)]));
            exit;
        });

        pcntl_signal(SIGINT, function () {
            ShopMonitorDataModel::query()->insert($this->insertData);
            $this->info(vsprintf('[%s] 触发SIGINT 入库 %s 条',
                [date('Y-m-d H:i:s'), count($this->insertData)]));
            exit;
        });
    }

}

